// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: fengyang

#include <algorithm>
#include <ctime>  // strptime
#include <experimental/filesystem>
#include <memory>
#include <regex>
#include <set>
#include <string>
#include <tuple>
#include <utility>
#include <vector>

#include <cxxopts.hpp>
#include "orio/io/base.h"
#include "orio/io/concat.h"
#include "orio/record/topic.h"

struct FlowFile {
  std::experimental::filesystem::path path;
  std::string topic_string;
  int64_t timestamp;
};

inline bool operator<(const FlowFile& lhs, const FlowFile& rhs) {
  return std::tie(lhs.timestamp, lhs.topic_string, lhs.path) <
         std::tie(rhs.timestamp, rhs.topic_string, rhs.path);
}

void PrintFLowFiles(std::vector<FlowFile>& flow_files) {
  for (const auto& f : flow_files) {
    std::cout << f.path.string() << std::endl;
  }
}

std::map<std::string, std::vector<FlowFile>> GetFileList(
   const std::string& root_dir) {
  std::map<std::string, std::vector<FlowFile>> out;

  for (const auto& entry : std::experimental::filesystem::directory_iterator(root_dir)) {
    //                                            012345678901234
    std::string filename =
        entry.path()
            .filename();  // e.g. binance_rest_depth_2.queue-0030ai8kg8.20190203-000000Z.gz
    std::istringstream iss(filename);
    std::string queue;
    std::string time;
    FlowFile file;
    file.path = entry.path();

    if (!(getline(iss, file.topic_string, '.') && getline(iss, queue, '.') &&
          getline(iss, time, '.')) ||
        time.size() != 16) {
      throw std::runtime_error("ill-formatted filename: " + file.path.string());
    }

    tm tm2{};
    if (strptime(time.c_str(), "%Y%m%d-%H%M%S%Z", &tm2) == nullptr) {
      throw std::runtime_error("ill-formatted filename: " + file.path.string());
    }
    time_t t = timegm(&tm2);
    file.timestamp = static_cast<int64_t>(t * 1'000'000'000LL);

    std::vector<FlowFile>& file_list = out[file.topic_string];

    file_list.push_back(file);
  }

  for (auto & [topic, file_list] : out) {
    (void)topic;
    std::sort(file_list.begin(), file_list.end());
  }

  return out;
}

int main(int argc, char* argv[]) {
  cxxopts::Options opt("Time revert checker", "General settting");
  opt.add_options()("root_dir", "root dir of rar feed", cxxopts::value<std::string>());
  opt.add_options()("topic", "topic to check", cxxopts::value<std::string>());
  auto setting = opt.parse(argc, argv);
  std::string root_dir = setting["root_dir"].as<std::string>();
  std::string topic_filter = setting["topic"].as<std::string>();

  std::map<std::string, std::vector<FlowFile>> topic_files= GetFileList(root_dir);
  int max_record_size = 10485760;
  std::unique_ptr<char[]> buf(new char[max_record_size]);
  auto* record = reinterpret_cast<orio::record::TopicRecord*>(buf.get());
  for (auto const& [topic, file_list] : topic_files) {
    if (!topic_filter.empty()) {
      if (topic_filter != topic) {
        continue;
      }
    }
    std::vector<std::experimental::filesystem::path> paths;
    for (auto const& file : file_list) {
      paths.push_back(file.path);
    }
    std::unique_ptr<orio::record::ConcatTopicReader> reader(
          new orio::record::ConcatTopicReader(paths, max_record_size, false));
    int64_t last_timestamp = 0;
    while (reader->Read(record)) {
      if (record->timestamp < last_timestamp) {
        std::cout << "Time invert, last " << last_timestamp << ", now " << record->timestamp << std::endl;

      }
      last_timestamp = record->timestamp;
    }

    std::cout << topic << " check done" << std::endl;
  }
}
